// bootstrap/kafka.go
package bootstrap

import (
	"github.com/Shopify/sarama"
	"github.com/amitshekhariitbhu/go-backend-clean-architecture/kafka"
)

type Kafka struct {
	Producer sarama.SyncProducer
	Consumer sarama.Consumer
}

func NewKafkaConfig(env *Env) *kafka.Config {
	return kafka.NewConfig(
		[]string{env.KafkaBrokers},
		env.KafkaUsername,
		env.KafkaPassword,
		env.KafkaTopic,
	)
}
func NewKafka(env *Env) *kafka.Config {
	return kafka.NewConfig(
		[]string{env.KafkaBrokers},
		env.KafkaUsername,
		env.KafkaPassword,
		env.KafkaTopic,
	)
}
func NewKafkaConnection(env *Env) *Kafka {
	config := sarama.NewConfig()
	config.Producer.Return.Successes = true
	config.Version = sarama.V2_0_0_0 // Adjust version as needed

	// If authentication is required
	if env.KafkaUsername != "" && env.KafkaPassword != "" {
		config.Net.SASL.Enable = true
		config.Net.SASL.User = env.KafkaUsername
		config.Net.SASL.Password = env.KafkaPassword
	}

	brokers := []string{env.KafkaBrokers} // Parse comma-separated if needed

	producer, err := sarama.NewSyncProducer(brokers, config)
	if err != nil {
		panic(err)
	}

	consumer, err := sarama.NewConsumer(brokers, config)
	if err != nil {
		panic(err)
	}

	return &Kafka{
		Producer: producer,
		Consumer: consumer,
	}
}

func (k *Kafka) Close() {
	if k.Producer != nil {
		k.Producer.Close()
	}
	if k.Consumer != nil {
		k.Consumer.Close()
	}
}
